package com.gao.api.gate.quick.proto.http;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.unix.Errors;
import io.netty.handler.codec.haproxy.HAProxyMessage;
import io.netty.handler.codec.http.*;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
import org.apache.commons.lang3.StringUtils;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.ListenableFuture;
import org.asynchttpclient.Response;
import org.asynchttpclient.request.body.multipart.ByteArrayPart;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class ClientRequestReceiver extends ChannelDuplexHandler {

    private io.netty.handler.codec.http.HttpRequest originalRequest;
    private HttpRequest request;
    private HttpRequestContext httpRequestContext;

    private static final Logger LOG = LoggerFactory.getLogger(ClientRequestReceiver.class);
    private static final String SCHEME_HTTP = "http";
    private static final String SCHEME_HTTPS = "https";
    public static final AttributeKey<Boolean> ATTR_LAST_CONTENT_RECEIVED = AttributeKey.newInstance("_last_content_received");


    public static boolean isLastContentReceivedForChannel(Channel ch) {
        Boolean value = ch.attr(ATTR_LAST_CONTENT_RECEIVED).get();
        return value == null ? false : value.booleanValue();
    }

    @Override
    public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {

        // Flag that we have now received the LastContent for this request from the client.
        // This is needed for ClientResponseReceiver to know whether it's yet safe to start writing
        // a response to the client channel.
        if (msg instanceof LastHttpContent) {
            ctx.channel().attr(ATTR_LAST_CONTENT_RECEIVED).set(Boolean.TRUE);
           // HttpBody requestBody = request.getRequestBody();
            /*LastHttpContent lastHttpContent = (LastHttpContent) msg;
            requestBody.loading(lastHttpContent);*/
           // requestBody.finishBufferedBodyIfIncomplete();
           // byte[] bytes = requestBody.bodyToBytes();


        }

        if (msg instanceof io.netty.handler.codec.http.HttpRequest) {
            originalRequest = (io.netty.handler.codec.http.HttpRequest) msg;
            if ("options".equalsIgnoreCase(originalRequest.method().name())) {
                final ChannelFuture f = ctx.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK));
                f.addListener((s) -> {
                    if (! s.isSuccess()) {
                        throw new RuntimeException("Failed while writing 100-continue response");
                    }
                });
            }else {
                //处理无效的 HTTP 请求。
                if (originalRequest.decoderResult().isFailure()) {

                } /*else if (request.isHasBody() && originalRequest. > zuulRequest.getMaxBodySize()) {

            }*/
                handleExpect100Continue(ctx, originalRequest);

                // 包装HttpRequest
                request = buildNewHttpRequest(originalRequest, ctx);

                //Send the request down the filter pipeline
                ctx.fireChannelRead(request);
            }
        }
        else if (msg instanceof HttpContent) {
            if (request != null) {
                HttpBody requestBody= null;
                if(request.getRequestBody() == null){
                    requestBody = new HttpBody();
                }else {
                    requestBody = request.getRequestBody();
                }
                HttpContent httpContent = (HttpContent) msg;

                requestBody.loading(httpContent);
                request.setRequestBody(requestBody);
           /*     byte[] bytes = requestBody.bodyToBytes();
                ByteBuf content = httpContent.content();
                int i = content.readableBytes();
                byte[] bytes1 = new byte[i];
                Headers headers = request.getHeaders();

                content.readBytes(bytes1);
                DefaultAsyncHttpClient defaultAsyncHttpClient = new DefaultAsyncHttpClient();
                ListenableFuture<Response> get1 = defaultAsyncHttpClient
                        .preparePost( "http://localhost:8081/hello")
                        .addHeader("Content-Type","application/json")
                        .setBody(bytes1)
                        .execute();
                Response response = get1.get();
                HttpHeaders headers1 = response.getHeaders();
                byte[] bytes2 = response.getResponseBody().getBytes();
                ctx.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(bytes2)));*/
               // ctx.writeAndFlush()
                ctx.fireChannelRead(msg);
            } else {
                //We already sent response for this request, these are laggard request body chunks that are still arriving
                ReferenceCountUtil.release(msg);
            }
        }
        else if (msg instanceof HAProxyMessage) {
            // do nothing, should already be handled by ElbProxyProtocolHandler
            LOG.debug("Received HAProxyMessage for Proxy Protocol IP: {}", ((HAProxyMessage) msg).sourceAddress());
            //ReferenceCountUtil.release(msg);
        }
        else {
            LOG.debug("Received unrecognized message type. " + msg.getClass().getName());
           // ReferenceCountUtil.release(msg);
        }
    }

    private HttpRequest buildNewHttpRequest(io.netty.handler.codec.http.HttpRequest originalRequest
            , ChannelHandlerContext ctx) {

        final Channel channel = ctx.channel();

        HttpRequest httpRequest = new HttpRequest(
                originalRequest.protocolVersion().toString(),
                originalRequest.method().toString(),
                originalRequest.uri()
        );
        // 包装请求参数
        httpRequest.setParams(copyQueryParams(originalRequest));
        // 包装请求头
        HttpHeaders headers = originalRequest.headers();
        httpRequest.setHeaders(Headers.build(headers));
        String val = originalRequest.headers().get("Transfer-Encoding");
        httpRequest.setHasBody(StringUtils.isNotBlank(val)&&"chunked".equalsIgnoreCase(val));
        return httpRequest;
    }

    private void handleExpect100Continue(ChannelHandlerContext ctx, io.netty.handler.codec.http.HttpRequest originalRequest) {
        if (HttpUtil.is100ContinueExpected(originalRequest)) {
            final ChannelFuture f = ctx.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE));
            f.addListener((s) -> {
                if (! s.isSuccess()) {
                    throw new RuntimeException("Failed while writing 100-continue response");
                }
            });
            // Remove the Expect: 100-Continue header from request as we don't want to proxy it downstream.
            originalRequest.headers().remove(HttpHeaderNames.EXPECT);
            request.getHeaders().remove(HttpHeaderNames.EXPECT.toString());
        }
    }


    public HttpParams copyQueryParams(final io.netty.handler.codec.http.HttpRequest originalRequest) {
        final String uri = originalRequest.uri();
        int queryStart = uri.indexOf('?');
        final String query = queryStart == -1 ? null : uri.substring(queryStart + 1);
        return HttpParams.parse(query);
    }


    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (msg instanceof HttpResponse) {
            promise.addListener((future) -> {
                if (! future.isSuccess()) {
                    fireWriteError("response headers", future.cause(), ctx);
                }
            });
            super.write(ctx, msg, promise);
        }
        else if (msg instanceof HttpContent) {
            promise.addListener((future) -> {
                if (! future.isSuccess())  {
                    fireWriteError("response content", future.cause(), ctx);
                }
            });
            super.write(ctx, msg, promise);
        }
        else {
            //should never happen
           // ReferenceCountUtil.release(msg);
          //  throw new ZuulException("Attempt to write invalid content type to client: "+msg.getClass().getSimpleName(), true);
        }
    }

    private void fireWriteError(String requestPart, Throwable cause, ChannelHandlerContext ctx) throws Exception {

        final String errMesg = String.format("Error writing %s to client", requestPart);

        if (cause instanceof java.nio.channels.ClosedChannelException ||
                cause instanceof Errors.NativeIoException) {
            LOG.info(errMesg + " - client connection is closed.");
           /* if (zuulRequest != null) {
                zuulRequest.getContext().cancel();
                StatusCategoryUtils.storeStatusCategoryIfNotAlreadyFailure(zuulRequest.getContext(), ZuulStatusCategory.FAILURE_CLIENT_CANCELLED);
            }*/
        }
        else {
            LOG.error(errMesg, cause);
           // ctx.fireExceptionCaught(new ZuulException(cause, errMesg, true));
        }
    }

}